﻿#region License
/* ---------------------------------------------------------------- *
 *
 * This file is part of the Xcoordination Application Space
 * ("XcoAppSpace") http://www.xcoordination.com
 *
 * Copyright (C) 2009 Xcoordination http://www.xcoordination.com
 *
 * XcoAppSpace is free software; you can redistribute it and/or
 * modify it under the terms of version 2 of the GNU General
 * Public License as published by the Free Software Foundation.
 *
 * XcoAppSpace is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program. If not, see http://www.gnu.org/licenses/
 * or write to the Free Software Foundation, Inc., 51 Franklin
 * Street, Fifth Floor, Boston, MA 02110-1301, USA.
 *
 * ---------------------------------------------------------------- */
#endregion

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Microsoft.Ccr.Core;
using XcoAppSpaces.Contracts.Exceptions;
using XcoAppSpaces.Contracts.Logging;

#if !PocketPC
using System.Linq.Expressions;
#endif

namespace XcoAppSpaces.Core
{
	/// <summary>
	/// The XcoPublisher manages a list of subscriptions in form of ports. It handles the
	/// processing of Subscribe and Unsubscribe messages, and the publishing messages
	/// to all subscribed ports.
	/// Is the base class for all generic XcoPublisher classes.
	/// </summary>
	public class XcoPublisher
	{
		internal class PublisherTask
		{
			public Subscription Subscription { get; set; }
			public Guid PortID { get; set; }
			public object Message { get; set; }
			public Type MessageType { get; set; }
		}

		internal class Subscription
		{
			public Subscribe Subscriber { get; set; }
			public Port<PublisherTask> PublishProcessorPort { get; set; }
		}

		private readonly IXcoLogger log = XcoLoggerFactory.CreateInstance(typeof(XcoPublisher));
		private readonly Dictionary<Type, Dictionary<Guid, Subscription>> subscriptions = new Dictionary<Type, Dictionary<Guid, Subscription>>();
		private readonly DispatcherQueue publisherQueue = new DispatcherQueue();
		//private Port<Exception> exceptionPort = new Port<Exception>();
		//private Causality c = null;
				
		internal XcoPublisher(params Type[] types)
		{
			//c = new Causality("xcopublisher", Guid.NewGuid(), exceptionPort, null);
			foreach (var type in types)
			{
				subscriptions.Add(type, new Dictionary<Guid, Subscription>());
			}
		}

		internal void ProcessSubscribe(Subscribe subscribe)
		{
			if (subscribe.Port == null)
				throw new XcoSubscriptionException("Subscription Port must not be null!");
			lock (subscriptions)
			{
				if (!subscriptions.ContainsKey(subscribe.PortType))
					throw new XcoSubscriptionException("Error subscribing to message: The type of message (" + subscribe.PortType + ") is unknown.");
				
				//local ports are recognized by instance, remote ports by their guid
				var portInfo = subscribe.Port.GetPortInfo();
				var publishProcessorPort = new Port<PublisherTask>();
				PortReceiveHelpers.InstallSequentialReceiver(publishProcessorPort, publisherQueue, ProcessPublisherTask);
				log.Debug("XcoPublisher: Subscribing port (" + (portInfo.IsRemotePort ? (portInfo.RemoteAddress + " - " +  portInfo.RemotePortID) : "local") + ") to type " + subscribe.PortType);
				var subscription = new Subscription { Subscriber = subscribe, PublishProcessorPort = publishProcessorPort };
				if (portInfo.IsRemotePort)
					subscriptions[subscribe.PortType][portInfo.RemotePortID] = subscription;
				else
					subscriptions[subscribe.PortType][Guid.NewGuid()] = subscription; //just create a random guid for a local port
			}
		}

		internal void ProcessUnsubscribe(Unsubscribe unsubscribe)
		{
			if (unsubscribe.Port == null)
				return;
			lock (subscriptions)
			{
				if (!subscriptions.ContainsKey(unsubscribe.PortType)) 
					return;

				//local ports are recognized by instance, remote ports by their guid
				var portInfo = unsubscribe.Port.GetPortInfo();
				log.Debug("XcoPublisher: Unubscribing port (" + (portInfo.IsRemotePort ? portInfo.RemoteAddress : "local") + ") from type " + unsubscribe.PortType);
				if (portInfo.IsRemotePort)
				{
					subscriptions[unsubscribe.PortType].Remove(portInfo.RemotePortID);
				}
				else
				{
					foreach (var subscription in subscriptions[unsubscribe.PortType])
					{
						if (subscription.Value.Subscriber.Port == unsubscribe.Port)
						{
							subscriptions[unsubscribe.PortType].Remove(subscription.Key);
							break;
						}
					}
				}
			}
		}

		/// <summary>
		/// Tries to publish a message of unknown type. An exception is thrown if the type
		/// of message is not supported by the XcoPublisher instance.
		/// </summary>
		/// <param name="msg">The message to be published.</param>
		public void PublishUnknownType(object msg)
		{
			Publish(msg, msg.GetType());
		}

		/// <summary>
		/// Tries to publish a message with the given type. An exception is thrown if the type
		/// of message is not supported by the XcoPublisher instance.
		/// </summary>
		/// <param name="msg">The message to be published.</param>
		/// <typeparam name="T">The type under which the message should be published (msg must be assignable to this type).</typeparam>
		protected void PublishWithType<T>(T msg)
		{
			Publish(msg, typeof (T));
		}

		private void Publish(object msg, Type type)
		{
			lock (subscriptions)
			{
				//Console.WriteLine("Publishing Type: " + type);
				if (subscriptions.ContainsKey(type))
				{
					foreach (var subscription in subscriptions[type])
					{
						if (subscription.Value.Subscriber.MatchesFilter(msg))
						{
							//Causality c = new Causality("xcopublisher" +(new Random()).Next(), exceptionPort);
							//Dispatcher.AddCausality(c);
							subscription.Value.PublishProcessorPort.Post(new PublisherTask { Subscription = subscription.Value, Message = msg, MessageType = type, PortID = subscription.Key });
							//Dispatcher.RemoveCausality(c);
						}
					}
				}
				else
				{
					log.Warn("XcoPublisher: Error publishing message: The type of message to be published (" + type + ") is unknown.");
					throw new XcoPublishException("Error publishing message: The type of message to be published (" + type + ") is unknown.");
				}
			}
		}

		private void ProcessPublisherTask(PublisherTask task)
		{
			lock (subscriptions)
			{
				if (!subscriptions[task.MessageType].ContainsKey(task.PortID))
					return;
			}
			try
			{
				log.Debug("XcoPublisher: Publishing message to port: " + task.PortID);
				task.Subscription.Subscriber.Port.SendUnknownType(task.Message);
			}
			catch (XcoCommunicationException ex)
			{
				lock (subscriptions)
				{
					subscriptions[task.MessageType].Remove(task.PortID);
				}
				task.Subscription.PublishProcessorPort.Clear(); //no need to try sending leftover items when remote port is not reachable any more
				log.Info("XcoPublisher: Port removed that is no more available: " + task.PortID + ": " + ex.Message);
			}
			catch (Exception)
			{
			}
		}
	}

	#region XcoPublisher<T>
	/// <summary>
	/// The XcoPublisher is a worker extension that manages a list of subscriptions in 
	/// form of ports. It handles the processing of Subscribe and Unsubscribe messages, 
	/// and provides methods to publish messages to all subscribed ports.
	/// The the message types that can be published are defined by the XcoPublisher's
	/// generic types.
	/// </summary>
	/// <typeparam name="T">The type of message to be published.</typeparam>
	[XcoWorkerExtension]
	public class XcoPublisher<T> : XcoPublisher
	{
		/// <summary>
		/// Creates a new XcoPublisher.
		/// </summary>
		public XcoPublisher(): base(typeof(T))
		{
		}

		/// <summary>
		/// Publishes the given message to all subscribed ports.
		/// </summary>
		/// <param name="msg">The message to be published.</param>
		public void Publish(T msg)
		{
			this.PublishWithType<T>(msg);
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		public void Subscribe(Port<T> port)
		{
			this.ProcessSubscribe(new Subscribe<T>(port));
		}

#if !PocketPC && !SILVERLIGHT
		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		/// <param name="filterFunc">Predicate that can be used for filtering the published messages.</param>
		public void Subscribe(Port<T> port, Expression<Predicate<T>> filterFunc)
		{
			this.ProcessSubscribe(new Subscribe<T>(port, filterFunc));
		}
#endif

		/// <summary>
		/// Unsubscribes the given port from receiving published messages.
		/// </summary>
		/// <param name="port">The port that should be unsubscribed.</param>
		public void Unsubscribe(Port<T> port)
		{
			this.ProcessUnsubscribe(new Unsubscribe<T>(port));
		}

		[XcoConcurrent]
		private void ProcessSubscribe(Subscribe<T> msg)
		{
			base.ProcessSubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessUnsubscribe(Unsubscribe<T> msg)
		{
			base.ProcessUnsubscribe(msg);
		}
	}
	#endregion

	#region XcoPublisher<T0, T1>
	/// <summary>
	/// The XcoPublisher is a worker extension that manages a list of subscriptions in 
	/// form of ports. It handles the processing of Subscribe and Unsubscribe messages, 
	/// and provides methods to publish messages to all subscribed ports.
	/// The the message types that can be published are defined by the XcoPublisher's
	/// generic types.
	/// </summary>
	/// <typeparam name="T0">The first message type to be published.</typeparam>
	/// <typeparam name="T1">The second message type to be published.</typeparam>
	[XcoWorkerExtension]
	public class XcoPublisher<T0, T1> : XcoPublisher
	{
		/// <summary>
		/// Creates a new XcoPublisher.
		/// </summary>
		public XcoPublisher()
			: base(typeof(T0), typeof(T1))
		{
		}

		/// <summary>
		/// Publishes the given message to all subscribed ports.
		/// </summary>
		/// <param name="msg">The message to be published.</param>
		public void Publish(T0 msg)
		{
			this.PublishWithType<T0>(msg);
		}

		/// <summary>
		/// Publishes the given message to all subscribed ports.
		/// </summary>
		/// <param name="msg">The message to be published.</param>
		public void Publish(T1 msg)
		{
			this.PublishWithType<T1>(msg);
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		public void Subscribe(Port<T0> port)
		{
			this.ProcessSubscribe(new Subscribe<T0>(port));
		}



		/// <summary>
		/// Unsubscribes the given port from receiving published messages.
		/// </summary>
		/// <param name="port">The port that should be unsubscribed.</param>
		public void Unsubscribe(Port<T0> port)
		{
			this.ProcessUnsubscribe(new Unsubscribe<T0>(port));
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		public void Subscribe(Port<T1> port)
		{
			this.ProcessSubscribe(new Subscribe<T1>(port));
		}

		/// <summary>
		/// Unsubscribes the given port from receiving published messages.
		/// </summary>
		/// <param name="port">The port that should be unsubscribed.</param>
		public void Unsubscribe(Port<T1> port)
		{
			this.ProcessUnsubscribe(new Unsubscribe<T1>(port));
		}

#if !PocketPC && !SILVERLIGHT
		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		/// <param name="filterFunc">Predicate that can be used for filtering the published messages.</param>
		public void Subscribe(Port<T0> port, Expression<Predicate<T0>> filterFunc)
		{
			this.ProcessSubscribe(new Subscribe<T0>(port, filterFunc));
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		/// <param name="filterFunc">Predicate that can be used for filtering the published messages.</param>
		public void Subscribe(Port<T1> port, Expression<Predicate<T1>> filterFunc)
		{
			this.ProcessSubscribe(new Subscribe<T1>(port, filterFunc));
		}
#endif

		[XcoConcurrent]
		private void ProcessSubscribe(Subscribe<T0> msg)
		{
			base.ProcessSubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessUnsubscribe(Unsubscribe<T0> msg)
		{
			base.ProcessUnsubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessSubscribe(Subscribe<T1> msg)
		{
			base.ProcessSubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessUnsubscribe(Unsubscribe<T1> msg)
		{
			base.ProcessUnsubscribe(msg);
		}
	}
	#endregion

	#region XcoPublisher<T0, T1, T2>
	/// <summary>
	/// The XcoPublisher is a worker extension that manages a list of subscriptions in 
	/// form of ports. It handles the processing of Subscribe and Unsubscribe messages, 
	/// and provides methods to publish messages to all subscribed ports.
	/// The the message types that can be published are defined by the XcoPublisher's
	/// generic types.
	/// </summary>
	/// <typeparam name="T0">The first message type to be published.</typeparam>
	/// <typeparam name="T1">The second message type to be published.</typeparam>
	/// <typeparam name="T2">The third message type to be published.</typeparam>
	[XcoWorkerExtension]
	public class XcoPublisher<T0, T1, T2> : XcoPublisher
	{
		/// <summary>
		/// Creates a new XcoPublisher.
		/// </summary>
		public XcoPublisher()
			: base(typeof(T0), typeof(T1), typeof(T2))
		{
		}

		/// <summary>
		/// Publishes the given message to all subscribed ports.
		/// </summary>
		/// <param name="msg">The message to be published.</param>
		public void Publish(T0 msg)
		{
			this.PublishWithType<T0>(msg);
		}

		/// <summary>
		/// Publishes the given message to all subscribed ports.
		/// </summary>
		/// <param name="msg">The message to be published.</param>
		public void Publish(T1 msg)
		{
			this.PublishWithType<T1>(msg);
		}

		/// <summary>
		/// Publishes the given message to all subscribed ports.
		/// </summary>
		/// <param name="msg">The message to be published.</param>
		public void Publish(T2 msg)
		{
			this.PublishWithType<T2>(msg);
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		public void Subscribe(Port<T0> port)
		{
			this.ProcessSubscribe(new Subscribe<T0>(port));
		}

		/// <summary>
		/// Unsubscribes the given port from receiving published messages.
		/// </summary>
		/// <param name="port">The port that should be unsubscribed.</param>
		public void Unsubscribe(Port<T0> port)
		{
			this.ProcessUnsubscribe(new Unsubscribe<T0>(port));
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		public void Subscribe(Port<T1> port)
		{
			this.ProcessSubscribe(new Subscribe<T1>(port));
		}

		/// <summary>
		/// Unsubscribes the given port from receiving published messages.
		/// </summary>
		/// <param name="port">The port that should be unsubscribed.</param>
		public void Unsubscribe(Port<T1> port)
		{
			this.ProcessUnsubscribe(new Unsubscribe<T1>(port));
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		public void Subscribe(Port<T2> port)
		{
			this.ProcessSubscribe(new Subscribe<T2>(port));
		}

		/// <summary>
		/// Unsubscribes the given port from receiving published messages.
		/// </summary>
		/// <param name="port">The port that should be unsubscribed.</param>
		public void Unsubscribe(Port<T2> port)
		{
			this.ProcessUnsubscribe(new Unsubscribe<T2>(port));
		}

#if !PocketPC && !SILVERLIGHT
		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		/// <param name="filterFunc">Predicate that can be used for filtering the published messages.</param>
		public void Subscribe(Port<T0> port, Expression<Predicate<T0>> filterFunc)
		{
			this.ProcessSubscribe(new Subscribe<T0>(port, filterFunc));
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		/// <param name="filterFunc">Predicate that can be used for filtering the published messages.</param>
		public void Subscribe(Port<T1> port, Expression<Predicate<T1>> filterFunc)
		{
			this.ProcessSubscribe(new Subscribe<T1>(port, filterFunc));
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		/// <param name="filterFunc">Predicate that can be used for filtering the published messages.</param>
		public void Subscribe(Port<T2> port, Expression<Predicate<T2>> filterFunc)
		{
			this.ProcessSubscribe(new Subscribe<T2>(port, filterFunc));
		}
#endif

		[XcoConcurrent]
		private void ProcessSubscribe(Subscribe<T0> msg)
		{
			base.ProcessSubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessUnsubscribe(Unsubscribe<T0> msg)
		{
			base.ProcessUnsubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessSubscribe(Subscribe<T1> msg)
		{
			base.ProcessSubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessUnsubscribe(Unsubscribe<T1> msg)
		{
			base.ProcessUnsubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessSubscribe(Subscribe<T2> msg)
		{
			base.ProcessSubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessUnsubscribe(Unsubscribe<T2> msg)
		{
			base.ProcessUnsubscribe(msg);
		}
	}
	#endregion

	#region XcoPublisher<T0, T1, T2, T3>
	/// <summary>
	/// The XcoPublisher is a worker extension that manages a list of subscriptions in 
	/// form of ports. It handles the processing of Subscribe and Unsubscribe messages, 
	/// and provides methods to publish messages to all subscribed ports.
	/// The the message types that can be published are defined by the XcoPublisher's
	/// generic types.
	/// </summary>
	/// <typeparam name="T0">The first message type to be published.</typeparam>
	/// <typeparam name="T1">The second message type to be published.</typeparam>
	/// <typeparam name="T2">The third message type to be published.</typeparam>
	/// <typeparam name="T3">The fourth message type to be published.</typeparam>
	[XcoWorkerExtension]
	public class XcoPublisher<T0, T1, T2, T3> : XcoPublisher
	{
		/// <summary>
		/// Creates a new XcoPublisher.
		/// </summary>
		public XcoPublisher()
			: base(typeof(T0), typeof(T1), typeof(T2), typeof(T3))
		{
		}

		/// <summary>
		/// Publishes the given message to all subscribed ports.
		/// </summary>
		/// <param name="msg">The message to be published.</param>
		public void Publish(T0 msg)
		{
			this.PublishWithType<T0>(msg);
		}

		/// <summary>
		/// Publishes the given message to all subscribed ports.
		/// </summary>
		/// <param name="msg">The message to be published.</param>
		public void Publish(T1 msg)
		{
			this.PublishWithType<T1>(msg);
		}

		/// <summary>
		/// Publishes the given message to all subscribed ports.
		/// </summary>
		/// <param name="msg">The message to be published.</param>
		public void Publish(T2 msg)
		{
			this.PublishWithType<T2>(msg);
		}

		/// <summary>
		/// Publishes the given message to all subscribed ports.
		/// </summary>
		/// <param name="msg">The message to be published.</param>
		public void Publish(T3 msg)
		{
			this.PublishWithType<T3>(msg);
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		public void Subscribe(Port<T0> port)
		{
			this.ProcessSubscribe(new Subscribe<T0>(port));
		}

		/// <summary>
		/// Unsubscribes the given port from receiving published messages.
		/// </summary>
		/// <param name="port">The port that should be unsubscribed.</param>
		public void Unsubscribe(Port<T0> port)
		{
			this.ProcessUnsubscribe(new Unsubscribe<T0>(port));
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		public void Subscribe(Port<T1> port)
		{
			this.ProcessSubscribe(new Subscribe<T1>(port));
		}

		/// <summary>
		/// Unsubscribes the given port from receiving published messages.
		/// </summary>
		/// <param name="port">The port that should be unsubscribed.</param>
		public void Unsubscribe(Port<T1> port)
		{
			this.ProcessUnsubscribe(new Unsubscribe<T1>(port));
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		public void Subscribe(Port<T2> port)
		{
			this.ProcessSubscribe(new Subscribe<T2>(port));
		}

		/// <summary>
		/// Unsubscribes the given port from receiving published messages.
		/// </summary>
		/// <param name="port">The port that should be unsubscribed.</param>
		public void Unsubscribe(Port<T2> port)
		{
			this.ProcessUnsubscribe(new Unsubscribe<T2>(port));
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		public void Subscribe(Port<T3> port)
		{
			this.ProcessSubscribe(new Subscribe<T3>(port));
		}

		/// <summary>
		/// Unsubscribes the given port from receiving published messages.
		/// </summary>
		/// <param name="port">The port that should be unsubscribed.</param>
		public void Unsubscribe(Port<T3> port)
		{
			this.ProcessUnsubscribe(new Unsubscribe<T3>(port));
		}

#if !PocketPC && !SILVERLIGHT
		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		/// <param name="filterFunc">Predicate that can be used for filtering the published messages.</param>
		public void Subscribe(Port<T0> port, Expression<Predicate<T0>> filterFunc)
		{
			this.ProcessSubscribe(new Subscribe<T0>(port, filterFunc));
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		/// <param name="filterFunc">Predicate that can be used for filtering the published messages.</param>
		public void Subscribe(Port<T1> port, Expression<Predicate<T1>> filterFunc)
		{
			this.ProcessSubscribe(new Subscribe<T1>(port, filterFunc));
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		/// <param name="filterFunc">Predicate that can be used for filtering the published messages.</param>
		public void Subscribe(Port<T2> port, Expression<Predicate<T2>> filterFunc)
		{
			this.ProcessSubscribe(new Subscribe<T2>(port, filterFunc));
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		/// <param name="filterFunc">Predicate that can be used for filtering the published messages.</param>
		public void Subscribe(Port<T3> port, Expression<Predicate<T3>> filterFunc)
		{
			this.ProcessSubscribe(new Subscribe<T3>(port, filterFunc));
		}
#endif

		[XcoConcurrent]
		private void ProcessSubscribe(Subscribe<T0> msg)
		{
			base.ProcessSubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessUnsubscribe(Unsubscribe<T0> msg)
		{
			base.ProcessUnsubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessSubscribe(Subscribe<T1> msg)
		{
			base.ProcessSubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessUnsubscribe(Unsubscribe<T1> msg)
		{
			base.ProcessUnsubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessSubscribe(Subscribe<T2> msg)
		{
			base.ProcessSubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessUnsubscribe(Unsubscribe<T2> msg)
		{
			base.ProcessUnsubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessSubscribe(Subscribe<T3> msg)
		{
			base.ProcessSubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessUnsubscribe(Unsubscribe<T3> msg)
		{
			base.ProcessUnsubscribe(msg);
		}
	}
	#endregion

	#region XcoPublisher<T0, T1, T2, T3, T4>
	/// <summary>
	/// The XcoPublisher is a worker extension that manages a list of subscriptions in 
	/// form of ports. It handles the processing of Subscribe and Unsubscribe messages, 
	/// and provides methods to publish messages to all subscribed ports.
	/// The the message types that can be published are defined by the XcoPublisher's
	/// generic types.
	/// </summary>
	/// <typeparam name="T0">The first message type to be published.</typeparam>
	/// <typeparam name="T1">The second message type to be published.</typeparam>
	/// <typeparam name="T2">The third message type to be published.</typeparam>
	/// <typeparam name="T3">The fourth message type to be published.</typeparam>
	/// <typeparam name="T4">The fifth message type to be published.</typeparam>
	[XcoWorkerExtension]
	public class XcoPublisher<T0, T1, T2, T3, T4> : XcoPublisher
	{
		/// <summary>
		/// Creates a new XcoPublisher.
		/// </summary>
		public XcoPublisher()
			: base(typeof(T0), typeof(T1), typeof(T2), typeof(T3), typeof(T4))
		{
		}

		/// <summary>
		/// Publishes the given message to all subscribed ports.
		/// </summary>
		/// <param name="msg">The message to be published.</param>
		public void Publish(T0 msg)
		{
			this.PublishWithType<T0>(msg);
		}

		/// <summary>
		/// Publishes the given message to all subscribed ports.
		/// </summary>
		/// <param name="msg">The message to be published.</param>
		public void Publish(T1 msg)
		{
			this.PublishWithType<T1>(msg);
		}

		/// <summary>
		/// Publishes the given message to all subscribed ports.
		/// </summary>
		/// <param name="msg">The message to be published.</param>
		public void Publish(T2 msg)
		{
			this.PublishWithType<T2>(msg);
		}

		/// <summary>
		/// Publishes the given message to all subscribed ports.
		/// </summary>
		/// <param name="msg">The message to be published.</param>
		public void Publish(T3 msg)
		{
			this.PublishWithType<T3>(msg);
		}

		/// <summary>
		/// Publishes the given message to all subscribed ports.
		/// </summary>
		/// <param name="msg">The message to be published.</param>
		public void Publish(T4 msg)
		{
			this.PublishWithType<T4>(msg);
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		public void Subscribe(Port<T0> port)
		{
			this.ProcessSubscribe(new Subscribe<T0>(port));
		}

		/// <summary>
		/// Unsubscribes the given port from receiving published messages.
		/// </summary>
		/// <param name="port">The port that should be unsubscribed.</param>
		public void Unsubscribe(Port<T0> port)
		{
			this.ProcessUnsubscribe(new Unsubscribe<T0>(port));
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		public void Subscribe(Port<T1> port)
		{
			this.ProcessSubscribe(new Subscribe<T1>(port));
		}

		/// <summary>
		/// Unsubscribes the given port from receiving published messages.
		/// </summary>
		/// <param name="port">The port that should be unsubscribed.</param>
		public void Unsubscribe(Port<T1> port)
		{
			this.ProcessUnsubscribe(new Unsubscribe<T1>(port));
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		public void Subscribe(Port<T2> port)
		{
			this.ProcessSubscribe(new Subscribe<T2>(port));
		}

		/// <summary>
		/// Unsubscribes the given port from receiving published messages.
		/// </summary>
		/// <param name="port">The port that should be unsubscribed.</param>
		public void Unsubscribe(Port<T2> port)
		{
			this.ProcessUnsubscribe(new Unsubscribe<T2>(port));
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		public void Subscribe(Port<T3> port)
		{
			this.ProcessSubscribe(new Subscribe<T3>(port));
		}

		/// <summary>
		/// Unsubscribes the given port from receiving published messages.
		/// </summary>
		/// <param name="port">The port that should be unsubscribed.</param>
		public void Unsubscribe(Port<T3> port)
		{
			this.ProcessUnsubscribe(new Unsubscribe<T3>(port));
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		public void Subscribe(Port<T4> port)
		{
			this.ProcessSubscribe(new Subscribe<T4>(port));
		}

		/// <summary>
		/// Unsubscribes the given port from receiving published messages.
		/// </summary>
		/// <param name="port">The port that should be unsubscribed.</param>
		public void Unsubscribe(Port<T4> port)
		{
			this.ProcessUnsubscribe(new Unsubscribe<T4>(port));
		}

#if !PocketPC && !SILVERLIGHT
		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		/// <param name="filterFunc">Predicate that can be used for filtering the published messages.</param>
		public void Subscribe(Port<T0> port, Expression<Predicate<T0>> filterFunc)
		{
			this.ProcessSubscribe(new Subscribe<T0>(port, filterFunc));
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		/// <param name="filterFunc">Predicate that can be used for filtering the published messages.</param>
		public void Subscribe(Port<T1> port, Expression<Predicate<T1>> filterFunc)
		{
			this.ProcessSubscribe(new Subscribe<T1>(port, filterFunc));
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		/// <param name="filterFunc">Predicate that can be used for filtering the published messages.</param>
		public void Subscribe(Port<T2> port, Expression<Predicate<T2>> filterFunc)
		{
			this.ProcessSubscribe(new Subscribe<T2>(port, filterFunc));
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		/// <param name="filterFunc">Predicate that can be used for filtering the published messages.</param>
		public void Subscribe(Port<T3> port, Expression<Predicate<T3>> filterFunc)
		{
			this.ProcessSubscribe(new Subscribe<T3>(port, filterFunc));
		}

		/// <summary>
		/// Subscribes the given port to receive published messages of the given port message type.
		/// </summary>
		/// <param name="port">The port that should receive published messages.</param>
		/// <param name="filterFunc">Predicate that can be used for filtering the published messages.</param>
		public void Subscribe(Port<T4> port, Expression<Predicate<T4>> filterFunc)
		{
			this.ProcessSubscribe(new Subscribe<T4>(port, filterFunc));
		}
#endif

		[XcoConcurrent]
		private void ProcessSubscribe(Subscribe<T0> msg)
		{
			base.ProcessSubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessUnsubscribe(Unsubscribe<T0> msg)
		{
			base.ProcessUnsubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessSubscribe(Subscribe<T1> msg)
		{
			base.ProcessSubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessUnsubscribe(Unsubscribe<T1> msg)
		{
			base.ProcessUnsubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessSubscribe(Subscribe<T2> msg)
		{
			base.ProcessSubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessUnsubscribe(Unsubscribe<T2> msg)
		{
			base.ProcessUnsubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessSubscribe(Subscribe<T3> msg)
		{
			base.ProcessSubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessUnsubscribe(Unsubscribe<T3> msg)
		{
			base.ProcessUnsubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessSubscribe(Subscribe<T4> msg)
		{
			base.ProcessSubscribe(msg);
		}

		[XcoConcurrent]
		private void ProcessUnsubscribe(Unsubscribe<T4> msg)
		{
			base.ProcessUnsubscribe(msg);
		}
	}
	#endregion
}
